package com.aliware.tianchi;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;
/**
 * 服务端过滤器
 * 可选接口
 * 此类可以修改实现，不可以移动类或者修改包名
 * 用户可以在服务端拦截请求和响应,捕获 rpc 调用时产生、服务端返回的已知异常。
 */
@Activate(group = CommonConstants.PROVIDER)
public class TestServerFilter implements Filter, BaseFilter.Listener {
    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        //
        URL url = invoker.getUrl();
        String methodName = invocation.getMethodName();

        int max = ProviderManager.max.get();
        //主要记录每台机器针对某个方法的并发数量
        RpcStatus count = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName());
        if (max > 0) {
            long timeout = 30;
            long start = System.currentTimeMillis();
            long remain = timeout;
            int active = count.getActive();
            if (active >= max) {
                synchronized (count) {
                    //这个while循环是必要的，因为在一次wait结束后，可能线程调用已经结束了，腾出来consumer的空间
                    while ((active = count.getActive()) >= max) {
                        try {
                            count.wait(remain);
                        } catch (InterruptedException e) {

                        }
                        //如果wait方法被中断的话，remain这时候有可能大于0
                        //如果其中一个线程运行结束自后调用notify方法的话，也有可能remain大于0
                        long elapsed = System.currentTimeMillis() - start;
                        remain = timeout - elapsed;
                        if (remain <= 0) {
                            return null;
                        }
                    }
                }
            }
        }
        long begin = System.currentTimeMillis();
        boolean state =false;
        try {
            //调用开始和结束后增减并发数量
            RpcStatus.beginCount(url, methodName);
            ProviderManager.startTime();
            Result result = invoker.invoke(invocation);
            state = true;
            RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, true);
            return result;
        }  catch (Exception t) {
            RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, false);
            throw t;
        } finally {
            ProviderManager.endTime(System.currentTimeMillis() - begin, state);
            //这里很关键，因为一个调用完成后要通知正在等待执行的队列
            synchronized (count) {
                count.notify();
            }
        }

/*
        long begin = System.currentTimeMillis();
        boolean state =false;
        try {
            ProviderManager.startTime();
            Result result = invoker.invoke(invocation);
            state = true;
            return result;
        } catch (Exception e) {
            throw e;
        } finally {
            ProviderManager.endTime(System.currentTimeMillis() - begin, state);
        }
*/

/*        try {
            //调用开始和结束后增减并发数量
            long begin = System.currentTimeMillis();
            RpcStatus.beginCount(url, methodName);
            try {
                boolean state =false;
                try {
                    ProviderManager.startTime();
                    Result result = invoker.invoke(invocation);
                    state = true;
                    RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, true);
                    return result;
                } catch (Exception e) {
                    throw e;
                } finally {
                    ProviderManager.endTime(System.currentTimeMillis() - begin, state);
                }
            } catch (RuntimeException t) {
                RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, false);
                throw t;
            }
        } finally {
            //这里很关键，因为一个调用完成后要通知正在等待执行的队列
            if(max>0){
                synchronized (count) {
                    count.notify();
                }
            }
        }
        //*/

    }

    @Override
    public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
        ProviderInfo providerInfo = ProviderManager.getServerInfo();
        ProviderManager.getSuccessCountPerSecond().incrementAndGet();
        String str1 = String.valueOf(providerInfo.getAllActiveCount());
        long avg = providerInfo.getAllSpendTimeTotal().get() / providerInfo.getAllReqCount().get();
        String str2 = String.valueOf(avg);
        int len = ProviderManager.getQueue().size();
        if (len == 0) len = 1;
        long avgTimeDuringOneSecond = ProviderManager.getTotalTime().get() / len;
        String key = "";
        for (int i = 0; i < ProviderManager.BUCKETNUM; i++) {
            key = "count" + String.valueOf(i);
            appResponse.setAttachment(key, String.valueOf(find(i)));
        }
        appResponse.setAttachment("avgTimeDuringSecond", String.valueOf(avgTimeDuringOneSecond));
        appResponse.setAttachment("len", String.valueOf(len));
        appResponse.setAttachment("activate", str1);
        appResponse.setAttachment("avgSpendTime", str2);
        appResponse.setAttachment("max",String.valueOf(ProviderManager.max));
    }

    @Override
    public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
    }
    public long find(int index) {
        Bucket bucket = ProviderManager.getBuckets().get(index);
        double num = bucket.getCount().get() * 0.99;
        if (num == 0) return -1;
        int n = (int) num;
        for (int i = 0; i <= 199; i++) {
            n -= bucket.getAtomicLongArray().get(i);
            if (n <= 0) {
                return i + 1;
            }
        }
        return 0;
    }
}
